package org.dbstar.scheduler.control.db.job;

import java.io.File;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.List;
import java.util.Set;

import org.apache.commons.lang.StringUtils;
import org.dbstar.scheduler.beans.Job;
import org.dbstar.scheduler.control.JobController;
import org.dbstar.scheduler.control.JobStatusChangeListener;
import org.dbstar.scheduler.control.db.JdbcTemplateDao;
import org.dbstar.scheduler.control.exception.DaoException;
import org.dbstar.scheduler.identifier.Identifier;
import org.dbstar.scheduler.identifier.IdentifierFactory;
import org.dbstar.scheduler.identifier.IdentifierType;
import org.dbstar.scheduler.job.JobStatus;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.dao.DataAccessException;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.jdbc.core.PreparedStatementSetter;
import org.springframework.jdbc.core.ResultSetExtractor;


/**
 * 使用JdbcTemplate的JobController实现类。
 * 
 * @author 代波
 * @since 1.0.0
 * @version 1.0.0
 * @date 2012-4-28
 */
public class JdbcTemplateJobController extends JdbcTemplateDao implements JobController {
	private static final Logger LOGGER = LoggerFactory.getLogger(JdbcTemplateJobController.class);
	private static final String WEB_APP_ROOT_KEY = "job.scheduler.root";
	private static final String WEB_APP_ROOT_KEY_REPLACE = "${" + WEB_APP_ROOT_KEY + "}";

	private final IdentifierFactory factory;
	private final JobQueryAll queryAll;
	private final JobQueryByStatus queryByStatus;
	private final JobQueryByConf queryByConf;
	private final JobQueryByIdentifier queryByIdentifier;
	private final JobQueryFirstReady queryFirstReady;
	private final JobInsert insert;
	private final JobUpdate update;
	private final JobUpdateStatus updateStatus;
	private final File resultRootFile;
	private final long lastMillis;

	public JdbcTemplateJobController(JdbcTemplate jt, IdentifierFactory factory, String resultRoot, int lastHour) {
		super(jt);
		if (resultRoot.indexOf(WEB_APP_ROOT_KEY_REPLACE) >= 0) {
			String webRoot = System.getProperty(WEB_APP_ROOT_KEY);
			resultRoot = StringUtils.replace(resultRoot, WEB_APP_ROOT_KEY_REPLACE, webRoot == null ? "" : webRoot);
		}
		this.resultRootFile = new File(resultRoot);
		LOGGER.info("Use job result root dir: {}", resultRootFile.getAbsolutePath());

		this.factory = factory;
		this.queryAll = new JobQueryAll(jt, resultRootFile);
		this.queryByStatus = new JobQueryByStatus(jt, resultRootFile);
		this.queryByConf = new JobQueryByConf(jt, resultRootFile);
		this.queryByIdentifier = new JobQueryByIdentifier(jt, resultRootFile);
		this.queryFirstReady = new JobQueryFirstReady(jt, resultRootFile);
		this.insert = new JobInsert(jt);
		this.update = new JobUpdate(jt);
		this.updateStatus = new JobUpdateStatus(jt);
		this.lastMillis = lastHour * 3600000L;
	}

	@Override
	public List<Job> getAllJobs() throws DaoException {
		try {
			return queryAll.execute(System.currentTimeMillis() - lastMillis);
		} catch (Throwable e) {
			throw new DaoException(e);
		}
	}

	@Override
	public List<Job> getJobs(JobStatus jobStatus) throws DaoException {
		try {
			return queryByStatus.execute(jobStatus.name(), System.currentTimeMillis() - lastMillis);
		} catch (Throwable e) {
			throw new DaoException(e);
		}
	}

	@Override
	public List<Job> getJobs(Identifier confIdentifier) throws DaoException {
		checkConfIdentifier(confIdentifier);
		try {
			return queryByConf.execute(confIdentifier.getSystemIdentifier(), confIdentifier.getSequence());
		} catch (Throwable e) {
			throw new DaoException(e);
		}
	}

	@Override
	public Job getJob(Identifier jobIdentifier) throws DaoException {
		checkIdentifier(jobIdentifier);
		try {
			return queryByIdentifier.findObject(jobIdentifier.getSystemIdentifier(), jobIdentifier.getSequence());
		} catch (Throwable e) {
			throw new DaoException(e);
		}
	}

	@Override
	public Job getFirstReadyJob() throws DaoException {
		try {
			return queryFirstReady.findFirstReady();
		} catch (Throwable e) {
			throw new DaoException(e);
		}
	}

	@Override
	public Job getFirstReadyJob(Set<String> excludeResources) throws DaoException {
		throw new DaoException(new UnsupportedOperationException());
	}

	private void checkConfIdentifier(Identifier identifier) throws DaoException {
		if (identifier == null) {
			throw new DaoException("Identifier must not null.");
		}
		if (identifier.getIdentifierType() != IdentifierType.conf) {
			throw new DaoException("Identifier not a Job Configuration identifier: " + identifier.getStringIdentifier());
		}
	}

	private void checkIdentifier(Identifier identifier) throws DaoException {
		if (identifier == null) {
			throw new DaoException("Identifier must not null.");
		}
		if (identifier.getIdentifierType() != IdentifierType.job) {
			throw new DaoException("Identifier not a Job identifier: " + identifier.getStringIdentifier());
		}
	}

	private void checkJob(Job job, boolean update) throws DaoException {
		if (job.getJobConfIdentifier() == null) {
			throw new DaoException("Job Configuration's Identifier must set.");
		}
		if (update) {
			if (job.getIdentifier() == null) {
				throw new DaoException("Job's Identifier must set.");
			}
		}
	}

	@Override
	public Identifier addJob(Job job) throws DaoException {
		checkJob(job, false);
		try {
			Identifier identifier = factory.createIdentifier(IdentifierType.job);
			insert.insertJob(job, identifier);
			job.setIdentifier(identifier);
			return identifier;
		} catch (Throwable e) {
			throw new DaoException(e);
		}

	}

	@Override
	public void updateJob(Job job) throws DaoException {
		checkJob(job, true);
		try {
			if (update.updateJob(job) == 0) {
				throw new IllegalArgumentException("Job not exist for update: "
						+ job.getIdentifier().getStringIdentifier());
			}
		} catch (Throwable e) {
			throw new DaoException(e);
		}
	}

	@Override
	public void changeJobStatus(Identifier jobIdentifier, JobStatus newStatus) throws DaoException {
		checkIdentifier(jobIdentifier);
		try {
			if (updateStatus.updateStatus(jobIdentifier, newStatus) == 0) {
				throw new IllegalArgumentException("Job not exist for update: " + jobIdentifier.getStringIdentifier());
			}
		} catch (Throwable e) {
			throw new DaoException(e);
		}
	}

	@Override
	public void registJobStatusChangeListener(JobStatusChangeListener listener, Identifier[] identifiers,
			Identifier confIdentifiers[], JobStatus... jobStatus) {
		throw new UnsupportedOperationException();
	}

	@Override
	public void removeJobStatusChangeListener(JobStatusChangeListener listener) {
		throw new UnsupportedOperationException();
	}

	@Override
	public List<Job> searchJobs(final Identifier jobIdentifier, final Identifier confIdentifier, final String type,
			final JobStatus status, final Boolean succeed, final String submitter, final Long submitFrom,
			final Long submitTo, final int page, final int pageSize) throws DaoException {
		StringBuilder where = new StringBuilder();
		if (jobIdentifier != null) {
			addCondition(where, "SYSTEM_IDENTIFIER = ? AND SEQUENCE = ?");
		}
		if (confIdentifier != null) {
			addCondition(where, "CONF_SYSTEM_IDENTIFIER = ? AND CONF_SEQUENCE = ?");
		}
		if (type != null) {
			addCondition(where, "TYPE = ?");
		}
		if (status != null) {
			addCondition(where, "STATUS = ?");
		}
		if (succeed != null) {
			addCondition(where, "SUCCEED = ?");
		}
		if (submitter != null) {
			addCondition(where, "SUBMITTER = ?");
		}
		if (submitFrom != null) {
			addCondition(where, "SUBMIT_TIME >= ?");
		}
		if (submitTo != null) {
			addCondition(where, "SUBMIT_TIME < ?");
		}
		String sql = "SELECT * FROM SCHD_JOB" + where.toString()
				+ " ORDER BY SYSTEM_IDENTIFIER DESC, SEQUENCE DESC LIMIT ?, ?";
		try {
			return jt.query(sql, new PreparedStatementSetter() {
				@Override
				public void setValues(PreparedStatement ps) throws SQLException {
					int index = 1;
					if (jobIdentifier != null) {
						ps.setString(index++, jobIdentifier.getSystemIdentifier());
						ps.setInt(index++, jobIdentifier.getSequence());
					}
					if (confIdentifier != null) {
						ps.setString(index++, confIdentifier.getSystemIdentifier());
						ps.setInt(index++, confIdentifier.getSequence());
					}
					if (type != null) {
						ps.setString(index++, type);
					}
					if (status != null) {
						ps.setString(index++, status.name());
					}
					if (succeed != null) {
						ps.setBoolean(index++, succeed.booleanValue());
					}
					if (submitter != null) {
						ps.setString(index++, submitter);
					}
					if (submitFrom != null) {
						ps.setLong(index++, submitFrom.longValue());
					}
					if (submitTo != null) {
						ps.setLong(index++, submitTo.longValue());
					}
					ps.setInt(index++, page * pageSize);
					ps.setInt(index++, pageSize);
				}
			}, new JobQuery.JobRowMapper(resultRootFile));
		} catch (Throwable e) {
			throw new DaoException(e);
		}
	}

	@Override
	public int searchJobSize(final Identifier jobIdentifier, final Identifier confIdentifier, final String type,
			final JobStatus status, final Boolean succeed, final String submitter, final Long submitFrom,
			final Long submitTo) throws DaoException {
		StringBuilder where = new StringBuilder();
		if (jobIdentifier != null) {
			addCondition(where, "SYSTEM_IDENTIFIER = ? AND SEQUENCE = ?");
		}
		if (confIdentifier != null) {
			addCondition(where, "CONF_SYSTEM_IDENTIFIER = ? AND CONF_SEQUENCE = ?");
		}
		if (type != null) {
			addCondition(where, "TYPE = ?");
		}
		if (status != null) {
			addCondition(where, "STATUS = ?");
		}
		if (succeed != null) {
			addCondition(where, "SUCCEED = ?");
		}
		if (submitter != null) {
			addCondition(where, "SUBMITTER = ?");
		}
		if (submitFrom != null) {
			addCondition(where, "SUBMIT_TIME >= ?");
		}
		if (submitTo != null) {
			addCondition(where, "SUBMIT_TIME < ?");
		}
		String sql = "SELECT COUNT(1) FROM SCHD_JOB" + where.toString();
		try {
			return jt.query(sql, new PreparedStatementSetter() {
				@Override
				public void setValues(PreparedStatement ps) throws SQLException {
					int index = 1;
					if (jobIdentifier != null) {
						ps.setString(index++, jobIdentifier.getSystemIdentifier());
						ps.setInt(index++, jobIdentifier.getSequence());
					}
					if (confIdentifier != null) {
						ps.setString(index++, confIdentifier.getSystemIdentifier());
						ps.setInt(index++, confIdentifier.getSequence());
					}
					if (type != null) {
						ps.setString(index++, type);
					}
					if (status != null) {
						ps.setString(index++, status.name());
					}
					if (succeed != null) {
						ps.setBoolean(index++, succeed.booleanValue());
					}
					if (submitter != null) {
						ps.setString(index++, submitter);
					}
					if (submitFrom != null) {
						ps.setLong(index++, submitFrom.longValue());
					}
					if (submitTo != null) {
						ps.setLong(index++, submitTo.longValue());
					}
				}
			}, new ResultSetExtractor<Integer>() {
				@Override
				public Integer extractData(ResultSet rs) throws SQLException, DataAccessException {
					return rs.next() ? rs.getInt(1) : 0;
				}
			});
		} catch (Throwable e) {
			throw new DaoException(e);
		}
	}

	private void addCondition(StringBuilder sql, String condition) {
		if (sql.length() == 0) {
			sql.append(" WHERE ");
		} else {
			sql.append(" AND ");
		}
		sql.append(condition);
	}
}
